﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Net;
using System.Threading;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.Diagnostics;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;
using System.IO;
using Microsoft.WindowsAzure.Search;

namespace Query
{
    public class WorkerRole : RoleEntryPoint
    {
        private Dictionary<int, Docinfo> docinfoDic = new Dictionary<int, Docinfo>();
        private List<string> partitionList = new List<string>();

        public override void Run()
        {
            // account
            CloudStorageAccount indexAccount = CloudStorageAccount.FromConfigurationSetting("IndexConnectionString");

            // client
            CloudBlobClient indexBlobClient = indexAccount.CreateCloudBlobClient();
            CloudQueueClient indexQueueClient = indexAccount.CreateCloudQueueClient();

            // queue
            CloudQueue queryQueue = indexQueueClient.GetQueueReference("query");
            queryQueue.CreateIfNotExist();
            CloudQueue scoreQueue = indexQueueClient.GetQueueReference("score");
            scoreQueue.CreateIfNotExist();
            CloudQueue docinfoQueue = indexQueueClient.GetQueueReference("docinfo");
            docinfoQueue.CreateIfNotExist();

            // container
            CloudBlobContainer scoreContainer = indexBlobClient.GetContainerReference("score");
            scoreContainer.CreateIfNotExist();

            // docinfo table context
            DocinfoContext docinfoContext = new DocinfoContext(indexAccount.TableEndpoint.ToString(), indexAccount.Credentials);
            IndexContext indexContext = new IndexContext(indexAccount.TableEndpoint.ToString(), indexAccount.Credentials);

            while (true)
            {
                // docinfo message
                CloudQueueMessage m = docinfoQueue.GetMessage();
                if (m != null)
                {
                    Debug.WriteLine(m.AsString);
                    docinfoQueue.DeleteMessage(m);
                    foreach (Docinfo docinfo in docinfoContext.Select(m.AsString))
                    {
                        docinfoDic.Add(Convert.ToInt32(docinfo.RowKey), docinfo);
                    }
                    partitionList.Add(m.AsString);
                    Debug.WriteLine("********docinfo inialize complete***********"); 
                }

                // query message
                m = queryQueue.GetMessage();
                if (m != null)
                {
                    // <docNum, partition, msgID, term ...>
                    String[] message = Tokenizer.CommaTokenize(m.AsString);
                    int docNum = Convert.ToInt32(message[0]);
                    String partition = message[1];
                    if (!partitionList.Contains(partition))
                        continue;

                    Stopwatch watch = new Stopwatch();
                    watch.Start();

                    queryQueue.DeleteMessage(m);
                    String msgID = message[2];
                    List<String> queryTerms = new List<string>();
                    for (int i = 3; i < message.Length; i++)
                        queryTerms.Add(message[i]);
           
                    // <docID, score>
                    Dictionary<int, double> scoreDic = new Dictionary<int, double>();

                    foreach (String term in queryTerms)
                    {
                        try
                        {
                            var result = indexContext.Select(partition, term);
                            int docFreq = indexContext.getDocFreq(term);
                            double idf = Math.Log10((double)docNum / (double)docFreq);

                            foreach (Index index in result)
                            {
                                String[] postings = Tokenizer.CommaTokenize(index.posting);
                                for (int i = 0; i < postings.Length; i += 2)
                                {
                                    int docID = Convert.ToInt32(postings[i]);
                                    int termFreq = Convert.ToInt32(postings[i + 1]);

                                    if (scoreDic.ContainsKey(docID))
                                        scoreDic[docID] += idf * termFreq;
                                    else
                                        scoreDic.Add(docID, idf * termFreq);
                                }
                            }
                        }
                        catch (Exception e)
                        {
                            continue;
                        }
                    }

                    String msg;
                    if (scoreDic.Count != 0)
                    {
                        // <msgID, partition>
                        String scoreName = msgID + "," + partition;
                        CloudBlob scoreBlob = scoreContainer.GetBlobReference(scoreName);
                        BlobStream scoreWriter = scoreBlob.OpenWrite();

                        bool first = true;
                        foreach (KeyValuePair<int, double> pair in scoreDic)
                        {
                            if (!first)
                                WriteData("\n", scoreWriter);

                            Docinfo docinfo = docinfoDic[pair.Key];
                            // <url, score>
                            WriteData(docinfo.url + "," + pair.Value / (double)docinfo.length, scoreWriter);
                            first = false;
                        }
                        scoreWriter.Flush();
                        scoreWriter.Close();

                        // <msgID, partition>
                        msg = msgID + "," + partition;
                    }
                    else
                    {
                        msg = msgID + "," + partition + ",none";
                    }
                    CloudQueueMessage scoreMsg = new CloudQueueMessage(msg);
                    scoreQueue.AddMessage(scoreMsg);
                    watch.Stop();
                    Debug.WriteLine("Score Time: " + watch.ElapsedMilliseconds + " ms");

                }

                Thread.Sleep(2000);
            }
        }
        
        private void WriteData(String str, BlobStream writer)
        {
            byte[] buffer = System.Text.Encoding.ASCII.GetBytes(str);
            writer.Write(buffer, 0, buffer.Length);
        }

        public override bool OnStart()
        {
            // Set the maximum number of concurrent connections 
            ServicePointManager.DefaultConnectionLimit = 12;

            // For information on handling configuration changes
            // see the MSDN topic at http://go.microsoft.com/fwlink/?LinkId=166357.

            #region Setup CloudStorageAccount Configuration Setting Publisher

            CloudStorageAccount.SetConfigurationSettingPublisher(
                (configName, configSetter) =>
                {
                    // Provide the configSetter with the initial value
                    configSetter(RoleEnvironment.GetConfigurationSettingValue(configName));
                    RoleEnvironment.Changed += (s, arg) =>
                    {
                        if (arg.Changes.OfType<RoleEnvironmentConfigurationSettingChange>()
                            .Any((change) => (change.ConfigurationSettingName == configName)))
                        {
                            if (!configSetter(RoleEnvironment.GetConfigurationSettingValue(
                                configName)))
                                // In this case, the change to the storage account credentials
                                // in the service configuration is significant enough that the
                                // role needs to be recycled in order to use the latest settings.
                                // (for example, the endpoint has changed)
                                RoleEnvironment.RequestRecycle();
                        }
                    };
                });
            #endregion

            return base.OnStart();
        }
    }
}
